package cn._51doit.day07;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.io.RandomAccessFile;
import java.util.Arrays;
import java.util.Collections;
import java.util.UUID;

/**
 * @create: 2021-10-23 15:27
 * @author: 今晚打脑斧先森
 * @program: MySourceFunction
 * @Description:
 *    在Source中定义OperatorState，是不跟key绑定再一起的
 *
 *    OperatorState 只有 一种形式 就是ListState
 *
 *    如果想使用OperatorState必须实现一个CheckpointedFunction接口，重写两个方法initializeState和snapshotState
 *
 *    initializeState是用来初始化或恢复状态的
 *    每做一次checkpoint，会调用snapshotState方法 然后进行偏移量的更新
 *
 *    执行过程 :
 *    initializeState（一次） ->  open(一次) -> run(一次，run方法中有while循环) -> 每做一次checkpoint方实现了CheckpointedFunction的subtask会调用一次snapshotState
 **/
public class MySourceFunction extends RichParallelSourceFunction<String>implements CheckpointedFunction {
    private boolean flag = true ;
    private String path ;
    private long offset = 0 ;

    //使用transient修饰的变量,不参与序列化和反序列化
    private transient ListState<Long> listState ;

    public MySourceFunction(String path) {//指定文件路径,写死不太好,这样写的好处就是可以在客户端定义路径
        this.path = path;
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
//        System.out.println("initializeState method invoke %%%%");
        //在initiallizeState方法中初始化OperatorState和恢复 OperatorState
        ListStateDescriptor<Long> stateDescriptor = new ListStateDescriptor<>("记录偏移量", Long.class);
        listState = context.getOperatorStateStore().getListState(stateDescriptor);
        if (context.isRestored()) { //判断是否有数据
            for (Long offset : listState.get()) {
                //从状态中恢复偏移量,并且赋值给成员offset
                this.offset = offset;
            }
        }
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
//        System.out.println("snapshotState method invoke %%%%");
        //清除以前的状态
        listState.clear();
        //将最新的偏移量存储起来
        listState.add(offset);

        /**
         * 获取新的list集合,保存offset数据
         * 替换掉旧的list集合
         *  listState.update(Collections.singletonList(offset));
         *  或者
         *  listState.update(Arrays.asList(offset));
         */

    }

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
//        System.out.println("run method invoke ~~~~");
        int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();

        RandomAccessFile raf = new RandomAccessFile(path + "/" + indexOfThisSubtask + ".txt" ,"r");
        //seek到指定的偏移量
        raf.seek(offset);
        while (flag){
            String line = raf.readLine(); //读取数据,一定要放在这里
            if (line!=null){
                synchronized (ctx.getCheckpointLock()){
                    line=new String(line.getBytes("ISO-8859-1"),"utf-8"); //解决中文乱码问题
                    ctx.collect(indexOfThisSubtask+" : "+line);
                    //获取最新的偏移量,赋值给offset
                    offset=raf.getFilePointer(); //获取文件的指向,也就是偏移量
                }
            }else {
                Thread.sleep(1000); //没有数据就睡眠1秒,不能浪费cpu资源
            }
        }
    }

    @Override
    public void cancel() {
//        System.out.println("cancel method invoke ~~~~");
        flag=false;
    }
}
